package com.atguigu.realtime.func;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.realtime.beans.TableProcess;
import com.atguigu.realtime.common.GmallConfig;
import com.atguigu.realtime.utils.DimUtil;
import com.atguigu.realtime.utils.HbaseUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.hadoop.hbase.client.Connection;

/**
 * @author: 洛尘
 * @since: 2023-09-26 23:59
 * @description: 将维度流中的数据写到Hbase表中
 **/
public class DimSinkFunction extends RichSinkFunction<JSONObject> {
    private Connection conn;

    @Override
    public void open(Configuration parameters) throws Exception {
        conn= HbaseUtil.getHbaseConnection();
    }

    @Override
    public void close() throws Exception {
        HbaseUtil.closeHbaseConnection(conn);
    }

    @Override
    public void invoke(JSONObject jsonObj, Context context) throws Exception {
        // {"tm_name":"Redmi123","id":1,"type":"update","table_process":{"op":"r","sinkColumns":"id,tm_name","sinkFamily":"info","sinkRowKey":"id","sinkTable":"dim_base_trademark","sourceTable":"base_trademark"}}
        //获取对业务数据库中维度表的操作类型
        String type = jsonObj.getString("type");
        //获取当前维度的对应的配置信息
        TableProcess tableProcess = jsonObj.getObject("table_process", TableProcess.class);
        //获取操作的bhase表的名称
        String sinkTable = tableProcess.getSinkTable();
        //获取rowKey名称
        String sinkRowKey = tableProcess.getSinkRowKey();
        //获取rowKey的值
        String rowkey = jsonObj.getString(sinkRowKey);

        //判断操作类型
        if ("delete".equals(type)){
            //从hbase对应的维度表中删除一条数据
            System.out.println("从hbase的"+ GmallConfig.HBASE_NAMESPACE +":"+sinkTable+"删除"+rowkey+"数据");
            HbaseUtil.delRow(conn,GmallConfig.HBASE_NAMESPACE,sinkTable,rowkey);
        }else {
            //向hbase对应的维度表中put一条数据
            System.out.println("向hbase的"+ GmallConfig.HBASE_NAMESPACE +":"+sinkTable+"表中put一条"+rowkey+"数据");
            //获取列族
            String sinkFamily = tableProcess.getSinkFamily();
            //获取列名
            String[] columnNames = tableProcess.getSinkColumns().split(",");
           //?
            String[] columnValues= new String[columnNames.length];
            for (int i = 0; i < columnNames.length; i++) {
                columnValues[i]= jsonObj.getString(columnNames[i]);
            }
            HbaseUtil.putRow(conn,GmallConfig.HBASE_NAMESPACE,sinkTable,rowkey,sinkFamily
            ,columnNames,columnValues);
        }
        if ("delete".equals(type)||"update".equals(type)){
            DimUtil.deleteCache(sinkTable,rowkey);
        }
    }



}
